package com.leonzhangxf.client.invoke;

import com.leonzhangxf.client.register.Node;
import com.leonzhangxf.client.register.ServiceLoader;
import com.leonzhangxf.client.register.ZkServiceLoader;
import com.leonzhangxf.common.JSONSerializer;
import com.leonzhangxf.common.RpcEncoder;
import com.leonzhangxf.common.RpcRequest;
import com.leonzhangxf.service.IUserService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

/**
 * @author leonzhangxf
 */
@Slf4j
@Component
public class ClientStarter {

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
        Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private ZkServiceLoader zkServiceLoader;

    public ClientStarter() {
        zkServiceLoader = new ZkServiceLoader("127.0.0.1:2181");
    }

    public IUserService userService() throws Exception {
        // 开启定时清理对应的服务节点负载数据
        zkServiceLoader.scheduleCleanData(IUserService.class.getName());
        return createProxy(IUserService.class, zkServiceLoader);
    }

    public static <T> T createProxy(final Class<T> serviceClass, ZkServiceLoader serviceLoader) throws Exception {
        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
            new Class[]{serviceClass}, new DelegateInvocationHandler<>(serviceClass, serviceLoader));
    }

    public static class NodeHandler<T> {

        @Getter
        private final Class<T> serviceClass;

        @Getter
        private final Node node;

        @Getter
        private final UserClientHandler userClientHandler;

        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();

        public NodeHandler(Class<T> serviceClass, Node node) {
            this.serviceClass = serviceClass;
            this.node = node;
            this.userClientHandler = new UserClientHandler();
        }

        public void shutdown() {
            group.shutdownGracefully();
        }

        public void initClient() throws InterruptedException {
            //4)配置启动引导对象
            bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //设置编码
                        pipeline.addLast(new StringDecoder());
                        //添加自定义事件处理器
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandler);
                    }
                });

            //5)连接服务端
            bootstrap.connect(node.getIp(), node.getPort()).sync();
        }
    }

    public static class DelegateInvocationHandler<T> implements InvocationHandler, ServiceLoader.NodeListChangeCallback {

        private final Class<T> serviceClass;

        private final List<NodeHandler<T>> nodeHandlers;

        private final ZkServiceLoader serviceLoader;

        private final Object lock = new Object();

        public DelegateInvocationHandler(Class<T> serviceClass, ZkServiceLoader serviceLoader) throws Exception {
            this.serviceClass = serviceClass;
            this.serviceLoader = serviceLoader;

            List<Node> nodes = serviceLoader.loadNode(serviceClass.getName(), this);

            synchronized (lock) {
                this.nodeHandlers = nodes.stream()
                    .map(node -> new NodeHandler<>(serviceClass, node))
                    .collect(Collectors.toList());
                this.nodeHandlers.forEach(handler -> {
                    try {
                        handler.initClient();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
        }

        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            //1)初始化客户端cliet
            RpcRequest rpcRequest = new RpcRequest();
            rpcRequest.setClassName(serviceClass.getName());
            rpcRequest.setMethodName(method.getName());
            rpcRequest.setParameters(args);
            rpcRequest.setParameterTypes(method.getParameterTypes());
            rpcRequest.setRequestId(UUID.randomUUID().toString());

            NodeHandler<T> nodeHandler;
            Node chooseNode;
            synchronized (lock) {
                log.info("当前节点有: " + nodeHandlers.stream().map(NodeHandler::getNode).collect(Collectors.toList()));
                // 负载获取节点
                chooseNode = serviceLoader.choose(serviceClass.getName());
                if (nodeHandlers.isEmpty() || null == chooseNode) {
                    nodeHandler = null;
                } else {
                    nodeHandler = nodeHandlers.stream()
                        .filter(handler -> handler.getNode().equals(chooseNode))
                        .findAny().orElse(null);
                }
                log.info("选择的节点: " + (null == nodeHandler ? null : nodeHandler.getNode()));
            }
            if (null == nodeHandler) {
                log.error("没有可用节点");
                return null;
            }
            UserClientHandler userClientHandler = nodeHandler.getUserClientHandler();
            //2)给UserClientHandler 设置param参数
            userClientHandler.setParam(rpcRequest);

            //3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
            //4)return 结果
            Future<Object> future = null;
            try {
                // 耗时记录
                long start = System.currentTimeMillis();
                future = executorService.submit(userClientHandler);
                Object result = future.get(2000, TimeUnit.MILLISECONDS);
                long end = System.currentTimeMillis();
                serviceLoader.updateLoadData(serviceClass.getName(), chooseNode, end - start);
                return result;
            } catch (TimeoutException ex) {
                ex.printStackTrace();
                future.cancel(true);
                // 暂时移出有问题的节点
                nodeHandlers.remove(nodeHandler);
                return null;
            }
        }

        void reCollectNodesHandlers(List<Node> nodes) {
            synchronized (lock) {
                // 删除没有的
                List<NodeHandler<T>> needRemove = nodeHandlers.stream()
                    .filter(handler -> !nodes.contains(handler.getNode())).collect(Collectors.toList());
                needRemove.forEach(NodeHandler::shutdown);
                nodeHandlers.removeAll(needRemove);

                // 新增新的
                List<Node> currentNodes = nodeHandlers.stream()
                    .map(NodeHandler::getNode).collect(Collectors.toList());
                List<Node> needCreate = Optional.ofNullable(nodes).orElse(Collections.emptyList()).stream()
                    .filter(node -> !currentNodes.contains(node)).collect(Collectors.toList());

                List<NodeHandler<T>> newCreate = needCreate.stream()
                    .map(node -> new NodeHandler<>(serviceClass, node))
                    .collect(Collectors.toList());
                newCreate.forEach(handler -> {
                    try {
                        handler.initClient();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
                nodeHandlers.addAll(newCreate);
            }
        }

        @Override
        public void nodeListChanged(List<Node> nodes) {
            List<Node> nodesSource = getNodesFromRegistry();
            if (null == nodesSource) {
                nodesSource = nodes;
            }
            reCollectNodesHandlers(nodesSource);
        }

        private List<Node> getNodesFromRegistry() {
            List<Node> nodesSource = null;
            try {
                nodesSource = serviceLoader.loadNode(serviceClass.getName(), this);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return nodesSource;
        }
    }
}
